package org.raftkv.raft.core;

import org.raftkv.commons.Triple;
import org.raftkv.commons.Tuple;
import org.raftkv.raft.confchange.Changer;
import org.raftkv.raft.core.enums.CampaignType;
import org.raftkv.raft.core.enums.StateType;
import org.raftkv.raft.core.exceptions.ProposalDroppedException;
import org.raftkv.raft.core.log.RaftLog;
import org.raftkv.raft.core.node.Ready;
import org.raftkv.raft.core.node.SoftState;
import org.raftkv.raft.core.readonly.ReadIndexStatus;
import org.raftkv.raft.core.readonly.ReadOnly;
import org.raftkv.raft.core.readonly.ReadState;
import org.raftkv.raft.core.storage.exceptions.SnapshotTemporarilyUnavailableException;
import org.raftkv.raft.exceptions.RaftException;
import org.raftkv.raft.pb.ConfState;
import org.raftkv.raft.pb.Entry;
import org.raftkv.raft.pb.HardState;
import org.raftkv.raft.pb.IConfChange;
import org.raftkv.raft.pb.Message;
import org.raftkv.raft.pb.Snapshot;
import org.raftkv.raft.pb.enums.EntryType;
import org.raftkv.raft.pb.enums.MessageType;
import org.raftkv.raft.pb.impl.ConfChange;
import org.raftkv.raft.pb.impl.ConfChangeV2;
import org.raftkv.raft.quorum.VoteResult;
import org.raftkv.raft.tracker.Inflights;
import org.raftkv.raft.tracker.Progress;
import org.raftkv.raft.tracker.ProgressMap;
import org.raftkv.raft.tracker.ProgressTracker;
import org.raftkv.raft.utils.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;
import java.util.function.Supplier;

/**
 * Raft 实现
 *
 * @author wzy
 */
public class Raft implements StateMachine {

    public static final long NONE = 0;

    public static final long NO_LIMIT = Long.MAX_VALUE;

    public static final int READ_ONLY_SAFE = 0;

    public static final int READ_ONLY_LEASE_BASED = 1;

    private long id;

    private long term;

    private long vote;

    private List<ReadState> readStates = new ArrayList<>();

    private RaftLog raftLog;

    private long maxMsgSize;

    private long maxUncommittedSize;

    private ProgressTracker prs;

    private StateType state;

    private boolean isLearner;

    private static final Logger LOGGER = LoggerFactory.getLogger(Raft.class);

    private long lead;

    private long leadTransferee;

    private long pendingConfIndex;

    private List<Message> msgs;

    private ReadOnly readOnly;

    /**
     * 上一次选举超时过去的时间
     */
    private int electionElapsed;

    /**
     * 收到上一次心跳过去的时间
     */
    private int heartbeatElapsed;

    private boolean checkQuorum;

    private long uncommittedSize;

    private int heartbeatTimeout;

    private boolean preVote;

    /**
     * 是一个在 [electionTimeout, 2 * electionTimeout - 1] 范围内的随机值
     */
    private int randomizedElectionTimeout;

    private boolean disableProposalForwarding;

    private int electionTimeout;

    private TickFunction tick;

    private StepFunction step;

    private List<Message> pendingReadIndexMessages;

    /**
     * 获取一个 Raft 实例
     *
     * @param c 算法设置
     * @return Raft
     */
    public static Raft newRaft(Config c) {
        c.validate();

        RaftLog raftLog = RaftLog.newLogWithSize(c.getStorage(), c.getMaxCommittedSizePerReady());
        Tuple<HardState, ConfState> hardStateConfStateTuple = c.getStorage().initialState();
        HardState hs = hardStateConfStateTuple.getFirst();
        ConfState cs = hardStateConfStateTuple.getSecond();

        Raft r = new Raft();
        r.id = c.getId();
        r.lead = NONE;
        r.isLearner = false;
        r.raftLog = raftLog;
        r.maxMsgSize = c.getMaxSizePerMsg();
        r.maxUncommittedSize = c.getMaxUncommittedEntriesSize();
        r.prs = ProgressTracker.makeProgressTracker(c.getMaxInflightMsgs());
        r.electionTimeout = c.getElectionTick();
        r.heartbeatTimeout = c.getHeartbeatTick();
        r.checkQuorum = c.isCheckQuorum();
        r.preVote = c.isPreVote();
        r.readOnly = ReadOnly.newReadOnly(c.getReadOnlyOption());
        r.disableProposalForwarding = c.isDisableProposalForwarding();
        r.state = StateType.StateFollower;

        Tuple<org.raftkv.raft.tracker.Config, ProgressMap> temp = ConfState.restore(new Changer(r.prs, raftLog.lastIndex()), cs);
        org.raftkv.raft.tracker.Config cfg = temp.getFirst();
        ProgressMap prs = temp.getSecond();

        Utils.assertConfStatesEquivalent(cs, r.switchToConfig(cfg, prs));

        // 如果不是第一次启动而是从之前的数据进行恢复
        if (!HardState.isEmptyHardState(hs)) {
            r.loadState(hs);
        }

        if (c.getApplied() > 0) {
            r.raftLog.appliedTo(c.getApplied());
        }
        r.becomeFollower(r.term, NONE);

        List<String> nodesStrs = new ArrayList<>();
        for (Long n : r.prs.voterNodes()) {
            nodesStrs.add(String.format("%x", n));
        }


        LOGGER.info("newRaft {} [peers: [{}], term: {}, commit: {}, applied: {}, lastindex: {}, lastterm: {}]",
                r.id, String.join(",", nodesStrs), r.term, r.raftLog.getCommitted(),
                r.raftLog.getApplied(), r.raftLog.lastIndex(), r.raftLog.lastTerm());
        return r;
    }

    private static int numOfPendingConf(List<Entry> ents) {
        int n = 0;
        if (ents != null) {
            for (Entry ent : ents) {
                if (ent.getType() == EntryType.EntryConfChange || ent.getType() == EntryType.EntryConfChangeV2) {
                    n++;
                }
            }
        }
        return n;
    }

    private static void releasePendingReadIndexMessages(Raft r) {
        if (!r.committedEntryInCurrentTerm()) {
            String errMsg = "pending MsgReadIndex should be released only after first commit in current term";
            LOGGER.error(errMsg);
            throw new RaftException(errMsg);
        }

        List<Message> msgs = r.pendingReadIndexMessages;
        r.pendingReadIndexMessages = null;

        if (msgs != null) {
            for (Message m : msgs) {
                sendMsgReadIndexResponse(r, m);
            }
        }
    }

    private static void sendMsgReadIndexResponse(Raft r, Message m) {
        switch (r.readOnly.getOption()) {
            case READ_ONLY_SAFE:
                r.readOnly.addRequest(r.raftLog.getCommitted(), m);
                r.readOnly.recvAck(r.id, m.getEntries().get(0).getData());
                r.bcastHeartbeatWithCtx(m.getEntries().get(0).getData());
                break;
            case READ_ONLY_LEASE_BASED:
                Message resp = r.responseToReadIndexReq(m, r.raftLog.getCommitted());
                if (resp.getTo() != NONE) {
                    r.send(resp);
                }
                break;
        }
    }

    private Raft() {

    }

    public boolean hasLeader() {
        return this.lead != NONE;
    }

    public SoftState softState() {
        return new SoftState(lead, state);
    }

    public HardState hardState() {
        return new HardState(raftLog.getCommitted(), term, vote);
    }

    public void send(Message m) {
        if (m.getFrom() == NONE) {
            m.setFrom(id);
        }

        if (m.getType() == MessageType.MsgVote || m.getType() == MessageType.MsgVoteResp || m.getType() == MessageType.MsgPreVote
                || m.getType() == MessageType.MsgPreVoteResp) {
            if (m.getTerm() == 0) {
                String errMsg = String.format("term should be set when sending %s", m.getType());
                LOGGER.error(errMsg);
                throw new RaftException(errMsg);
            }
        } else {
            if (m.getTerm() != 0) {
                String errMsg = String.format("term should not be set when sending %s (was %d)", m.getType(), m.getTerm());
                LOGGER.error(errMsg);
                throw new RaftException(errMsg);
            }

            if (m.getType() != MessageType.MsgProp && m.getType() != MessageType.MsgReadIndex) {
                m.setTerm(this.term);
            }
        }
        if (msgs == null) {
            msgs = new ArrayList<>();
        }
        msgs.add(m);
    }

    private void sendAppend(Long to) {
        maybeSendAppend(to, true);
    }

    private boolean maybeSendAppend(long to, boolean sendIfEmpty) {
        Progress pr = this.prs.getProgress().getProgressMap().get(to);
        if (pr.isPaused()) {
            return false;
        }

        Message m = new Message();
        m.setTo(to);

        long term;
        List<Entry> ents;
        try {
            term = raftLog.term(pr.getNext() - 1);
            ents = raftLog.entries(((int) pr.getNext()), maxMsgSize);

            if ((ents == null || ents.size() == 0) && !sendIfEmpty) {
                return false;
            }

            m.setType(MessageType.MsgApp);
            m.setIndex(pr.getNext() - 1);
            m.setLogTerm(term);
            m.setEntries(ents);
            m.setCommit(this.raftLog.getCommitted());
            int n = m.getEntries() == null ? 0 : m.getEntries().size();
            if (n != 0) {
                switch (pr.getState()) {
                    case StateReplicate:
                        long last = m.getEntries().get(n - 1).getIndex();
                        pr.optimisticUpdate(last);
                        pr.getInflights().add(last);
                        break;
                    case StateProbe:
                        pr.setProbeSent(true);
                        break;
                    default:
                        String errMsg = String.format("%x is sending append in unhandled state %s", this.id, pr.getState());
                        LOGGER.error(errMsg);
                        throw new RaftException(errMsg);
                }
            }
        } catch (Throwable ex) {
            if (!pr.isRecentActive()) {
                LOGGER.debug("ignore sending snapshot to {} since it is not recently active", to);
                return false;
            }

            m.setType(MessageType.MsgSnap);
            Snapshot snapshot;
            try {
                snapshot = this.raftLog.snapshot();
            } catch (SnapshotTemporarilyUnavailableException e) {
                LOGGER.debug("{} failed to send snapshot to {} because snapshot is temporarily unavailable", id, to);

                return false;
            } catch (Exception e) {
                throw new RaftException(e);
            }

            if (Snapshot.isEmptySnap(snapshot)) {
                throw new RaftException("need non-empty snapshot");
            }
            m.setSnapShot(snapshot);
            long sindex = snapshot.getMetadata().getIndex();
            long sterm = snapshot.getMetadata().getTerm();
            LOGGER.debug("{} [firstindex: {}, commit: {}] sent snapshot[index: {}, term: {}] to {} [{}]",
                    this.id, this.raftLog.firstIndex(), this.raftLog.getCommitted(), sindex, sterm, to, pr);
            pr.becomeSnapshot(sindex);
            LOGGER.debug("{} paused sending replication messages to {} [{}]", this.id, to, pr);
        }

        send(m);
        return true;
    }

    public void sendHeartbeat(long to, byte[] ctx) {
        // commit index取需要发送过去的节点的match已经当前leader的commited中的较小值
        long commit = Math.min(this.prs.getProgress().getProgressMap().get(to).getMatch(), raftLog.getCommitted());
        send(Message.builder()
                .to(to)
                .type(MessageType.MsgHeartbeat)
                .commit(commit)
                .context(ctx)
                .build());
    }

    public void bcastAppend() {
        this.prs.visit((id, prs) -> {
            if (id == this.id) {
                return;
            }
            sendAppend(id);
        });
    }

    private void bcastHeartbeat() {
        String lastCtx = this.readOnly.lastPendingRequestCtx();
        if (lastCtx.length() == 0) {
            bcastHeartbeatWithCtx(null);
        } else {
            bcastHeartbeatWithCtx(lastCtx.getBytes());
        }
    }

    private void bcastHeartbeatWithCtx(byte[] ctx) {
        this.prs.visit((id, prs) -> {
            if (id == this.id) {
                return;
            }
            sendHeartbeat(id, ctx);
        });
    }

    public void advance(Ready rd) {
        reduceUncommittedSize(rd.getCommittedEntries());

        long newApplied = rd.appliedCursor();
        if (newApplied > 0) {
            long oldApplied = raftLog.getApplied();
            raftLog.appliedTo(newApplied);

            if (prs.getConfig().isAutoLeave() && oldApplied <= pendingConfIndex && newApplied >= pendingConfIndex
                    && state == StateType.StateLeader) {
                Entry ent = new Entry();
                ent.setType(EntryType.EntryConfChangeV2);
                ent.setData(null);

                if (!appendEntry(ent)) {
                    throw new RaftException("refused un-refusable auto-leaving ConfChangeV2");
                }
                pendingConfIndex = raftLog.lastIndex();
                LOGGER.info("initiating automatic transition out of joint configuration {}", prs.getConfig());
            }
        }

        if (rd.getEntries().size() > 0) {
            Entry e = rd.getEntries().get(rd.getEntries().size() - 1);
            raftLog.stableTo(e.getIndex(), e.getTerm());
        }

        if (Snapshot.isEmptySnap(rd.getSnapshot())) {
            raftLog.stableSnapTo(rd.getSnapshot().getMetadata().getIndex());
        }
    }

    /**
     * 尝试提交日志
     *
     * @return 是否提交成功
     */
    public boolean maybeCommit() {
        long mci = this.prs.committed();
        // 排列之后拿到中位数的 Match，因为如果这个位置的 Match 对应的 Term 也等于当前的 Term
        // 说明有过半的节点至少 commit 了 mci 这个索引的数据，这样 leader 就可以以这个索引进行 commit了
        return this.raftLog.maybeCommit(mci, this.term);
    }

    /**
     * 重置raft的状态
     *
     * @param term term
     */
    public void reset(long term) {
        if (this.term != term) {
            // 如果是新的任期，那么保存任期号，同时将投票节点置空
            this.term = term;
            this.vote = NONE;
        }
        lead = NONE;

        electionElapsed = 0;
        heartbeatElapsed = 0;
        resetRandomizedElectionTimeout();

        abortLeaderTransfer();

        prs.resetVotes();
        prs.visit((id, prs) -> {
            Progress pr = new Progress();
            pr.setMatch(0);
            pr.setNext(raftLog.lastIndex() + 1);
            pr.setInflights(Inflights.newInflights(this.prs.getMaxInflight()));
            pr.setLearner(prs.getProgressMap().get(id).isLearner());

            if (id == this.id) {
                pr.setMatch(raftLog.lastIndex());
            }
            prs.getProgressMap().put(id, pr);
        });

        pendingConfIndex = 0;
        uncommittedSize = 0;
        readOnly = new ReadOnly(readOnly.getOption());
    }

    public boolean appendEntry(Entry... es) {
        long li = raftLog.lastIndex();
        for (int i = 0; i < es.length; i++) {
            Entry entry = es[i];
            entry.setTerm(term);
            entry.setIndex(li + i + 1);
        }

        if (!increaseUncommittedSize(List.of(es))) {
            LOGGER.debug("{} appending new entries to log would exceed uncommitted entry size limit; dropping proposal", id);
            return false;
        }
        li = raftLog.append(Arrays.stream(es).toList());
        // 更新本节点的Next以及Match索引
        prs.getProgress().getProgressMap().get(id).maybeUpdate(li);
        // append之后，尝试一下是否可以进行commit
        maybeCommit();

        return true;
    }

    /**
     * follower以及candidate的tick函数，在r.electionTimeout之后被调用
     *
     * @return tickFunction
     */
    private TickFunction tickElection() {
        return () -> {
            ++electionElapsed;
            if (promotable() && pastElectionTimeout()) {
                // 如果可以被提升为 leader，同时选举时间也到了
                electionElapsed = 0;
                // 发送HUP消息是为了重新开始选举
                step(Message.builder()
                        .from(id)
                        .type(MessageType.MsgHup)
                        .build());
            }
        };
    }

    /**
     * tickHeartbeat 是 leader 的 tick 函数，用于探测 Follower 的心跳
     */
    private TickFunction tickHeartBeat() {
        return () -> {
            ++heartbeatElapsed;
            ++electionElapsed;
            if (electionElapsed >= electionTimeout) {
                // 如果超过了选举时间
                electionElapsed = 0;
                if (checkQuorum) {
                    step(Message.builder()
                            .from(id)
                            .type(MessageType.MsgCheckQuorum)
                            .build());
                }

                if (state == StateType.StateLeader && leadTransferee != NONE) {
                    abortLeaderTransfer();
                }
            }

            if (state != StateType.StateLeader) {
                return;
            }

            if (heartbeatElapsed >= heartbeatTimeout) {
                // 向集群中其他节点发送广播消息
                heartbeatElapsed = 0;
                // 尝试发送 MsgBeat 消息
                step(Message.builder()
                        .from(id)
                        .type(MessageType.MsgBeat)
                        .build());
            }
        };
    }

    public void becomeFollower(long term, long lead) {
        step = stepFollower();
        reset(term);
        this.lead = lead;
        this.tick = tickElection();
        this.state = StateType.StateFollower;
        LOGGER.info("{} became follower at term {}", id, term);
    }

    public void becomeCandidate() {
        if (state == StateType.StateLeader) {
            throw new RaftException("invalid transition [leader -> candidate]");
        }
        step = stepCandidate();
        // 因为进入candidate状态，意味着需要重新进行选举了，所以reset的时候传入的是Term+1
        reset(term + 1);
        tick = tickElection();
        // 给自己投票
        vote = id;
        state = StateType.StateCandidate;
        LOGGER.info("{} became candidate at term {}", id, term);
    }

    public void becomePreCandidate() {
        if (state == StateType.StateLeader) {
            throw new RaftException("invalid transition [leader -> pre-candidate]");
        }
        step = stepCandidate();
        prs.resetVotes();
        tick = tickElection();
        lead = NONE;
        state = StateType.StatePreCandidate;
        LOGGER.info("{} became pre-candidate at term {}", id, term);
    }

    public void becomeLeader() {
        if (state == StateType.StateFollower) {
            throw new RaftException("invalid transition [follower -> leader]");
        }
        step = stepLeader();
        reset(term);
        tick = tickHeartBeat();
        lead = id;
        state = StateType.StateLeader;
        prs.getProgress().getProgressMap().get(id).becomeReplicate();
        pendingConfIndex = raftLog.lastIndex();
        Entry emptyEnt = new Entry();
        emptyEnt.setData(null);

        if (!appendEntry(emptyEnt)) {
            String errMsg = "empty entry was dropped";
            LOGGER.error(errMsg);
            throw new RaftException(errMsg);
        }
        reduceUncommittedSize(List.of(emptyEnt));
        LOGGER.info("{} became leader at term {}", id, term);
    }

    private void hup(CampaignType t) {
        if (state == StateType.StateLeader) {
            LOGGER.debug("{} ignoring MsgHup because already leader", id);
            return;
        }

        if (!promotable()) {
            LOGGER.warn("{} is unpromotable and can not campaign", id);
            return;
        }

        List<Entry> ents;
        try {
            ents = raftLog.slice(raftLog.getApplied() + 1, raftLog.getCommitted() + 1, NO_LIMIT);
        } catch (Throwable e) {
            String errMsg = String.format("unexpected error getting unapplied entries (%s)", e);
            LOGGER.error(errMsg, e);
            throw new RaftException("", e);
        }

        int n = numOfPendingConf(ents);
        if (n != 0 && raftLog.getCommitted() > raftLog.getApplied()) {
            LOGGER.warn("{} cannot campaign at term {} since there are still {} pending configuration changes to apply",
                    id, term, n);
            return;
        }

        LOGGER.info("{} is starting a new election at term {}", id, term);
        campaignInternal(t);
    }

    public void campaignInternal(CampaignType t) {
        if (!promotable()) {
            LOGGER.warn("{} is unpromotable; campaign() should have been called", id);
        }

        long term = 0;
        MessageType voteMsg;
        if (t == CampaignType.CampaignPreElection) {
            becomePreCandidate();
            voteMsg = MessageType.MsgPreVote;
            term = this.term + 1;
        } else {
            becomeCandidate();
            voteMsg = MessageType.MsgVote;
            term = this.term;
        }

        Triple<Integer, Integer, VoteResult> triple = poll(id, Utils.voteRespMsgType(voteMsg), true);
        VoteResult res = triple.getThird();
        if (res == VoteResult.VoteWon) {
            if (t == CampaignType.CampaignPreElection) {
                campaignInternal(CampaignType.CampaignElection);
            } else {
                becomeLeader();
            }
            return;
        }

        List<Long> ids = new ArrayList<>(prs.getConfig().getVoters().ids());
        ids.sort(Long::compare);

        for (Long id : ids) {
            if (this.id == id) {
                continue;
            }
            LOGGER.info("{} [logterm: {}, index: {}] sent {} request to {} at term {}",
                    this.id, this.raftLog.lastTerm(), this.raftLog.lastIndex(), voteMsg, id, this.term);

            byte[] ctx = new byte[0];
            if (t == CampaignType.CampaignTransfer) {
                ctx = t.getDesc().getBytes();
            }
            send(new Message.MessageBuilder()
                    .term(term)
                    .to(id)
                    .type(voteMsg)
                    .index(raftLog.lastIndex())
                    .logTerm(raftLog.lastTerm())
                    .context(ctx)
                    .build());
        }
    }

    /**
     * 轮询集群中所有节点，返回一共有多少节点已经进行了投票
     *
     * @param id 投票id
     * @param t  投票类型
     * @param v  是否投票
     */
    private Triple<Integer, Integer, VoteResult> poll(long id, MessageType t, boolean v) {
        if (v) {
            LOGGER.info("{} received {} from {} at term {}", this.id, t, id, term);
        } else {
            LOGGER.info("{} received {} rejection from {} at term {}", this.id, t, id, term);
        }
        prs.recordVote(id, v);
        return prs.tallyVotes();
    }

    /**
     * Raft 状态机
     *
     * @param m 接收到的消息
     */
    @Override
    public void step(Message m) {
        // 来自本地的消息
        if (m.getTerm() == 0) {
            // do nothing
        } else if (m.getTerm() > term) {
            if (m.getType() == MessageType.MsgVote || m.getType() == MessageType.MsgPreVote) {
                boolean force = new String(m.getContext() == null ? new byte[0] : m.getContext()).equals(CampaignType.CampaignTransfer.getDesc());
                boolean inLease = checkQuorum && this.lead != NONE && electionElapsed < electionTimeout;
                if (!force && inLease) {
                    LOGGER.info("{} [logterm: {}, index: {}, vote: {}] ignored {} from {} [logterm: {}, index: {}] at term {}: lease is not expired (remaining ticks: {})",
                            id, raftLog.lastTerm(), raftLog.lastIndex(), vote, m.getType(), m.getFrom(), m.getLogTerm(), m.getIndex(), term, electionTimeout - electionElapsed);
                    return;
                }
            }

            if (m.getType() == MessageType.MsgPreVote) {
                ;
            } else if (m.getType() == MessageType.MsgPreVoteResp && !m.isReject()) {
                ;
            } else {
                LOGGER.info("{} [term: {}] received a {} message with higher term from {} [term: {}]",
                        id, term, m.getType(), m.getFrom(), m.getTerm());
                if (m.getType() == MessageType.MsgApp || m.getType() == MessageType.MsgHeartbeat || m.getType() == MessageType.MsgSnap) {
                    becomeFollower(m.getTerm(), m.getFrom());
                } else {
                    becomeFollower(m.getTerm(), NONE);
                }
            }
        } else if (m.getTerm() < term) {
            // 消息的 Term 小于节点自身的 Term，同时消息类型是心跳消息或者是 append 消息
            // 收到了一个节点发送过来的更小的 term 消息。这种情况可能是因为消息的网络延时导致，
            // 但是也可能因为该节点由于网络分区导致了它递增了term到一个新的任期。
            // 这种情况下该节点不能赢得一次选举，也不能使用旧的任期号重新再加入集群中。
            // 如果 checkQuorum 为 false，这种情况可以使用递增任期号应答来处理。
            // 但是如果 checkQuorum 为 true，此时收到了一个更小的 term 的节点发出的 HB 或者 APP 消息，
            // 于是应答一个 AppResp 消息，试图纠正它的状态
            if ((checkQuorum || preVote) && (m.getType() == MessageType.MsgHeartbeat || m.getType() == MessageType.MsgApp)) {
                send(Message.builder()
                        .to(m.getFrom())
                        .type(MessageType.MsgAppResp)
                        .build());
            } else if (m.getType() == MessageType.MsgPreVote) {
                LOGGER.info("{} [logterm: {}, index: {}, vote: {}] rejected {} from {} [logterm: {}, index: {}] at term {}",
                        id, raftLog.lastTerm(), raftLog.lastIndex(), vote, m.getType(), m.getFrom(), m.getLogTerm(), m.getIndex(), m.getTerm());
                send(Message.builder()
                        .to(m.getFrom())
                        .term(term)
                        .reject(true)
                        .type(MessageType.MsgPreVoteResp)
                        .build());
            } else {
                // 除了上面的情况以外，忽略任何term小于当前节点所在任期号的消息
                LOGGER.info("{} [term: {}] ignored a {} message with lower term from {} [term: {}]",
                        id, term, m.getType(), m.getFrom(), m.getTerm());
            }
            // 在消息的term小于当前节点的term时，不往下处理直接返回了
            return;
        }

        switch (m.getType()) {
            // 收到 HUP 消息，说明准备进行选举
            case MsgHup:
                if (preVote) {
                    hup(CampaignType.CampaignPreElection);
                } else {
                    hup(CampaignType.CampaignElection);
                }
                break;
            // 收到投票类的消息
            case MsgPreVote:
            case MsgVote:
                boolean canVote = vote == m.getFrom() ||
                        (vote == NONE && lead == NONE) ||
                        (m.getType() == MessageType.MsgPreVote && m.getTerm() > term);
                if (canVote && raftLog.isUpToDate(m.getIndex(), m.getLogTerm())) {
                    // 如果当前没有给任何节点投票（r.Vote == None）或者投票的节点term大于本节点的（m.Term > r.Term）
                    // 或者是之前已经投票的节点（r.Vote == m.From）
                    // 同时还满足该节点的消息是最新的（r.raftLog.isUpToDate(m.Index, m.LogTerm)），那么就接收这个节点的投票
                    LOGGER.info("{} [logterm: {}, index: {}, vote: {}] cast {} for {} [logterm: {}, index: {}] at term {}",
                            id, raftLog.lastTerm(), raftLog.lastIndex(), vote, m.getType(), m.getFrom(), m.getLogTerm(),
                            m.getIndex(), m.getTerm());
                    send(Message.builder()
                            .to(m.getFrom())
                            .term(m.getTerm())
                            .type(Utils.voteRespMsgType(m.getType()))
                            .build());
                    // 保存下来给哪个节点投票了
                    if (m.getType() == MessageType.MsgVote) {
                        electionElapsed = 0;
                        vote = m.getFrom();
                    }
                } else {
                    // 否则拒绝投票
                    LOGGER.info("{} [logterm: {}, index: {}, vote: {}] rejected {} from {} [logterm: {}, index: {}] at term {}",
                            id, raftLog.lastTerm(), raftLog.lastIndex(), vote, m.getType(), m.getFrom(), m.getLogTerm(),
                            m.getIndex(), term);
                    send(Message.builder()
                            .to(m.getFrom())
                            .term(term)
                            .type(Utils.voteRespMsgType(m.getType()))
                            .reject(true)
                            .build());
                }
                break;
            // 其他情况下进入各种状态下自己定制的状态机函数
            default:
                step.step(this, m);
        }
    }

    public StepFunction stepLeader() {
        return (raft, m) -> {
            switch (m.getType()) {
                case MsgBeat:
                    // 广播HB消息
                    bcastHeartbeat();
                    return;
                case MsgCheckQuorum:
                    Progress pr = prs.getProgress().getProgressMap().get(id);
                    if (pr != null) {
                        pr.setRecentActive(true);
                    }
                    if (!prs.quorumActive()) {
                        LOGGER.warn("{} stepped down to follower since quorum is not active", id);
                        becomeFollower(term, NONE);
                    }
                    prs.visit((id, prs) -> {
                        if (id != this.id) {
                            prs.getProgressMap().get(id).setRecentActive(false);
                        }
                    });
                    return;
                case MsgProp:
                    // 不能提交空数据
                    if (m.getEntries().size() == 0) {
                        String errMsg = String.format("%x stepped empty MsgProp", id);
                        LOGGER.error(errMsg);
                        throw new RaftException(errMsg);
                    }
                    // 检查是否在集群中
                    if (!prs.getProgress().getProgressMap().containsKey(id)) {
                        // 这里检查本节点是否还在集群以内，如果已经不在集群中了，不处理该消息直接返回。
                        // 这种情况出现在本节点已经通过配置变化被移除出了集群的场景。
                        throw new ProposalDroppedException();
                    }

                    if (leadTransferee != NONE) {
                        LOGGER.debug("{} [term {}] transfer leadership to {} is in progress; dropping proposal",
                                id, term, leadTransferee);
                        throw new ProposalDroppedException();
                    }

                    for (int i = 0; i < m.getEntries().size(); i++) {
                        Entry e = m.getEntries().get(i);
                        IConfChange cc = null;
                        if (e.getType() == EntryType.EntryConfChange) {
                            ConfChange ccc = new ConfChange();
                            ccc.unmarshal(e.getData());
                            cc = ccc;
                        } else if (e.getType() == EntryType.EntryConfChangeV2) {
                            ConfChangeV2 ccc = new ConfChangeV2();
                            ccc.unmarshal(e.getData());
                            cc = ccc;
                        }

                        if (cc != null) {
                            boolean alreadyPending = pendingConfIndex > raftLog.getApplied();
                            boolean alreadyJoint = (prs.getConfig().getVoters().CONFIGS[1] == null ? 0 : prs.getConfig().getVoters().CONFIGS[1].size()) > 0;
                            boolean wantsLeaveJoint = cc.asV2().getChanges().size() == 0;

                            String refused = "";
                            if (alreadyPending) {
                                refused = String.format("possible unapplied conf change at index %d (applied to %d)",
                                        pendingConfIndex, raftLog.getApplied());
                            } else if (alreadyJoint && !wantsLeaveJoint) {
                                refused = "must transition out of joint config first";
                            } else if (!alreadyJoint && wantsLeaveJoint) {
                                refused = "not in joint state; refusing empty conf change";
                            }

                            if (!"".equals(refused)) {
                                LOGGER.info("{} ignoring conf change {} at config {}: {}",
                                        id, cc, prs.getConfig(), refused);
                                m.getEntries().set(i, new Entry(EntryType.EntryNormal));
                            } else {
                                pendingConfIndex = raftLog.lastIndex() + i + 1;
                            }
                        }
                    }

                    // 添加数据到log中
                    if (!appendEntry(m.getEntries().toArray(new Entry[0]))) {
                        throw new ProposalDroppedException();
                    }
                    // 向集群其他节点广播append消息
                    bcastAppend();
                    return;
                case MsgReadIndex:
                    if (prs.isSingleton()) {
                        Message resp = responseToReadIndexReq(m, raftLog.getCommitted());
                        if (resp.getTo() != NONE) {
                            send(resp);
                        }
                        return;
                    }

                    if (!committedEntryInCurrentTerm()) {
                        pendingReadIndexMessages.add(m);
                        return;
                    }

                    sendMsgReadIndexResponse(this, m);
                    return;
            }
            // 检查消息发送者当前是否在集群中
            Progress pr = prs.getProgress().getProgressMap().get(m.getFrom());
            if (pr == null) {
                LOGGER.debug("{} no progress available for {}", id, m.getFrom());
                return;
            }

            switch (m.getType()) {
                case MsgAppResp:
                    // 置位该节点当前是活跃的
                    pr.setRecentActive(true);
                    if (m.isReject()) {
                        // 如果拒绝了append消息，说明term、index不匹配
                        LOGGER.debug("{} received MsgAppResp(rejected, hint: (index {}, term {})) from {} for index {}",
                                id, m.getRejectHint(), m.getLogTerm(), m.getFrom(), m.getIndex());
                        // rejectHint 带来的是拒绝该app请求的节点，其最大日志的索引
                        long nextProbeIdx = m.getRejectHint();
                        if (m.getLogTerm() > 0) {
                            nextProbeIdx = raftLog.findConflictByTerm(m.getRejectHint(), m.getLogTerm());
                        }
                        if (pr.maybeDecrTo(m.getIndex(), nextProbeIdx)) {
                            LOGGER.debug("{} decreased progress of {} to [{}]", id, m.getFrom(), pr);
                            if (pr.getState() == org.raftkv.raft.tracker.enums.StateType.StateReplicate) {
                                pr.becomeProbe();
                            }
                            sendAppend(m.getFrom());
                        }
                    } else {// 通过该append请求
                        boolean oldPaused = pr.isPaused();
                        // 如果该节点的索引发生了更新
                        if (pr.maybeUpdate(m.getIndex())) {
                            if (pr.getState() == org.raftkv.raft.tracker.enums.StateType.StateProbe) {// 如果当前该节点在探测状态，切换到可以接收副本状态
                                pr.becomeReplicate();
                            } else if (pr.getState() == org.raftkv.raft.tracker.enums.StateType.StateSnapshot
                                    && pr.getMatch() >= pr.getPendingSnapshot()) {
                                // 如果当前该接在在接受快照状态，而且已经快照数据同步完成了
                                LOGGER.debug("{} recovered from needing snapshot, resumed sending replication messages to {} [{}]",
                                        id, m.getFrom(), pr);
                                pr.becomeProbe();
                                pr.becomeReplicate();
                            } else if (pr.getState() == org.raftkv.raft.tracker.enums.StateType.StateReplicate) {
                                pr.getInflights().freeLE(m.getIndex());
                            }

                            if (maybeCommit()) {
                                // 如果可以commit日志，那么广播append消息
                                releasePendingReadIndexMessages(this);
                                bcastAppend();
                            } else if (oldPaused) {
                                // 如果该节点之前状态是暂停，继续发送append消息给它
                                sendAppend(m.getFrom());
                            }

                            while (maybeSendAppend(m.getFrom(), false)) {
                                ;
                            }
                            if (m.getFrom() == leadTransferee && pr.getMatch() == raftLog.lastIndex()) {
                                LOGGER.info("{} sent MsgTimeoutNow to {} after received MsgAppResp", id, m.getFrom());
                                sendTimeoutNow(m.getFrom());
                            }
                        }
                    }
                    break;
                case MsgHeartbeatResp:
                    // 该节点当前处于活跃状态
                    pr.setRecentActive(true);
                    // 这里调用resume是因为当前可能处于probe状态，
                    // 而这个状态在两个heartbeat消息的间隔期只能收一条同步日志消息，因此在收到HB消息时就停止pause标记
                    pr.setProbeSent(false);

                    if (pr.getState() == org.raftkv.raft.tracker.enums.StateType.StateReplicate && pr.getInflights().full()) {
                        pr.getInflights().freeFirstOne();
                    }

                    /// 该节点的match节点小于当前最大日志索引，可能已经过期了，尝试添加日志
                    if (pr.getMatch() < raftLog.lastIndex()) {
                        sendAppend(m.getFrom());
                    }

                    if (readOnly.getOption() != READ_ONLY_SAFE || (m.getContext() == null || m.getContext().length == 0)) {
                        return;
                    }

                    // 小于集群半数以上就返回不往下走了
                    if (prs.getConfig().getVoters().voteResult(readOnly.recvAck(m.getFrom(), m.getContext())) != VoteResult.VoteWon) {
                        return;
                    }

                    // 调用advance函数尝试丢弃已经被确认的read index状态
                    List<ReadIndexStatus> rss = readOnly.advance(m);
                    for (ReadIndexStatus rs : rss) {
                        Message resp = responseToReadIndexReq(rs.getReq(), rs.getIndex());
                        if (resp.getTo() != NONE) {
                            send(resp);
                        }
                    }
                    break;
                case MsgSnapStatus:
                    if (pr.getState() != org.raftkv.raft.tracker.enums.StateType.StateSnapshot)
                        return;
                    if (!m.isReject()) {
                        pr.becomeProbe();
                        LOGGER.debug("{} snapshot succeeded, resumed sending replication messages to {} [{}]",
                                id, m.getFrom(), pr);
                    } else {
                        pr.setPendingSnapshot(0);
                        pr.becomeProbe();
                        LOGGER.debug("{} snapshot failed, resumed sending replication messages to {} [{}]",
                                id, m.getFrom(), pr);
                    }
                    pr.setProbeSent(true);
                    break;
                case MsgUnreachable:
                    if (pr.getState() == org.raftkv.raft.tracker.enums.StateType.StateReplicate) {
                        pr.becomeProbe();
                    }
                    LOGGER.debug("{} failed to send message to {} because it is unreachable [{}]",
                            id, m.getFrom(), pr);
                    break;
                case MsgTransferLeader:
                    if (pr.isLearner()) {
                        LOGGER.debug("{} is learner. Ignored transferring leadership", id);
                        return;
                    }
                    long leadTransferee = m.getFrom();
                    long lastLeadTransferee = this.leadTransferee;
                    if (lastLeadTransferee != NONE) {
                        if (lastLeadTransferee == leadTransferee) {
                            LOGGER.info("{} [term {}] transfer leadership to {} is in progress, ignores request to same node {}",
                                    id, term, leadTransferee, leadTransferee);
                            return;
                        }
                        abortLeaderTransfer();
                        LOGGER.info("{} [term {}] abort previous transferring leadership to {}",
                                id, term, lastLeadTransferee);
                    }

                    if (leadTransferee == id) {
                        LOGGER.debug("{} is already leader. Ignored transferring leadership to self", id);
                        return;
                    }

                    LOGGER.info("{} [term {}] starts to transfer leadership to {}", id, term, leadTransferee);
                    electionElapsed = 0;
                    this.leadTransferee = leadTransferee;

                    if (pr.getMatch() == raftLog.lastIndex()) {
                        sendTimeoutNow(leadTransferee);
                        LOGGER.info("{} sends MsgTimeoutNow to {} immediately as {} already has up-to-date log",
                                id, leadTransferee, leadTransferee);
                    } else {
                        sendAppend(leadTransferee);
                    }
            }
        };
    }

    public StepFunction stepCandidate() {
        return (r, m) -> {
            MessageType myVoteRespType;
            if (r.state == StateType.StatePreCandidate) {
                myVoteRespType = MessageType.MsgPreVoteResp;
            } else {
                myVoteRespType = MessageType.MsgVoteResp;
            }
            if (m.getType() == MessageType.MsgProp) {// 当前没有leader，所以忽略掉提交的消息
                LOGGER.info("{} no leader at term {}; dropping proposal", id, term);
                throw new ProposalDroppedException();
            } else if (m.getType() == MessageType.MsgApp) {// 收到append消息，说明集群中已经有leader，转换为follower
                becomeFollower(m.getTerm(), m.getFrom());
                handleAppendEntries(m);
            } else if (m.getType() == MessageType.MsgHeartbeat) {// 收到HB消息，说明集群已经有leader，转换为follower
                becomeFollower(m.getTerm(), m.getFrom());
                handleHeartbeat(m);
            } else if (m.getType() == MessageType.MsgSnap) {// 收到快照消息，说明集群已经有leader，转换为follower
                becomeFollower(m.getTerm(), m.getFrom());
                handleSnapshot(m);
            } else if (m.getType() == myVoteRespType) {// 计算当前集群中有多少节点给自己投了票
                Triple<Integer, Integer, VoteResult> triple = poll(m.getFrom(), m.getType(), !m.isReject());
                Integer gr = triple.getFirst();
                Integer rj = triple.getSecond();
                VoteResult res = triple.getThird();
                LOGGER.info("{} has received {} {} votes and {} vote rejections",
                        id, gr, m.getType(), rj);
                switch (res) {
                    case VoteWon:
                        if (r.state == StateType.StatePreCandidate) {
                            campaignInternal(CampaignType.CampaignElection);
                        } else {
                            becomeLeader();
                            bcastAppend();
                        }
                        break;
                    case VoteLost:
                        becomeFollower(term, NONE);
                }
            } else if (m.getType() == MessageType.MsgTimeoutNow) {
                LOGGER.debug("{} [term {} state {}] ignored MsgTimeoutNow from {}",
                        id, term, state, m.getFrom());
            }
        };
    }

    public StepFunction stepFollower() {
        return (raft, m) -> {
            switch (m.getType()) {
                case MsgProp:
                    if (lead == NONE) {
                        // 没有leader则提交失败，忽略
                        LOGGER.info("{} no leader at term {}; dropping proposal", id, term);
                        throw new ProposalDroppedException();
                    } else if (disableProposalForwarding) {
                        LOGGER.info("{} not forwarding to leader {} at term {}; dropping proposal", id, lead, term);
                        throw new ProposalDroppedException();
                    }
                    // 向leader转发
                    m.setTo(lead);
                    send(m);
                    break;
                case MsgApp:
                    // append消息
                    // 收到leader的app消息，重置选举tick计时器，因为这样证明leader还存活
                    electionElapsed = 0;
                    lead = m.getFrom();
                    handleAppendEntries(m);
                    break;
                case MsgHeartbeat:
                    electionElapsed = 0;
                    lead = m.getFrom();
                    handleHeartbeat(m);
                    break;
                case MsgSnap:
                    electionElapsed = 0;
                    lead = m.getFrom();
                    handleSnapshot(m);
                    break;
                case MsgTransferLeader:
                    if (lead == NONE) {
                        LOGGER.info("{} no leader at term {}; dropping leader transfer msg", id, term);
                        return;
                    }
                    m.setTo(lead);
                    send(m);
                    break;
                case MsgTimeoutNow:
                    LOGGER.info("{} [term {}] received MsgTimeoutNow from {} and starts an election to get leadership.",
                            id, term, m.getFrom());
                    hup(CampaignType.CampaignTransfer);
                    break;
                case MsgReadIndex:
                    if (lead == NONE) {
                        LOGGER.info("{} no leader at term {}; dropping index reading msg", id,
                                term);
                        return;
                    }
                    m.setTo(lead);
                    send(m);
                    break;
                case MsgReadIndexResp:
                    if (m.getEntries().size() != 1) {
                        String errMsg = String.format("%x invalid format of MsgReadIndexResp from %x, entries count: %d", id, m.getFrom(), m.getEntries().size());
                        LOGGER.error(errMsg);
                        throw new RaftException(errMsg);
                    }
                    readStates.add(new ReadState(m.getIndex(), m.getEntries().get(0).getData()));
            }
        };
    }

    public void handleAppendEntries(Message m) {
        // 先检查消息消息的合法性
        if (m.getIndex() < raftLog.getCommitted()) {
            send(Message.builder()
                    .to(m.getFrom())
                    .type(MessageType.MsgAppResp)
                    .index(raftLog.getCommitted())
                    .build());
            return;
        }

        Tuple<Long, Boolean> t = raftLog.maybeAppend(m.getIndex(), m.getLogTerm(), m.getCommit(), m.getEntries());
        Long mlastIndex = t.getFirst();
        Boolean ok = t.getSecond();
        if (ok) {
            // 添加成功，返回的index是添加成功之后的最大index
            send(Message.builder()
                    .to(m.getFrom())
                    .type(MessageType.MsgAppResp)
                    .index(mlastIndex)
                    .build());
        } else {
            LOGGER.debug("{} [logterm: {}, index: {}] rejected msgApp [logterm: {}, index: {}] from {}",
                    id, raftLog.zeroTermOnErrCompacted(m.getIndex()), m.getIndex(), m.getLogTerm(),
                    m.getIndex(), m.getFrom());
            long hintIndex = Math.min(m.getIndex(), raftLog.lastIndex());
            hintIndex = raftLog.findConflictByTerm(hintIndex, m.getLogTerm());
            long hintTerm;
            try {
                hintTerm = raftLog.term(hintIndex);
            } catch (Throwable e) {
                String errMsg = String.format("term(%d) must be valid, but got %s", hintIndex, e);
                throw new RaftException(errMsg);
            }
            send(Message.builder()
                    .to(m.getFrom())
                    .type(MessageType.MsgAppResp)
                    .index(m.getIndex())
                    .reject(true)
                    .rejectHint(hintIndex)
                    .logTerm(hintTerm)
                    .build()
            );
        }
    }

    /**
     * 处理心跳消息
     *
     * @param m m
     */
    public void handleHeartbeat(Message m) {
        raftLog.commitTo(m.getCommit());
        send(Message.builder()
                .to(m.getFrom())
                .type(MessageType.MsgHeartbeatResp)
                // 要把HB消息带过来的context原样返回
                .context(m.getContext())
                .build());
    }

    private void handleSnapshot(Message m) {
        long sindex = m.getSnapShot().getMetadata().getIndex();
        long sterm = m.getSnapShot().getMetadata().getTerm();
        // 注意这里成功与失败，只是返回的Index参数不同
        if (restore(m.getSnapShot())) {
            LOGGER.info("{} [commit: {}] restored snapshot [index: {}, term: {}]",
                    id, raftLog.getCommitted(), sindex, sterm);
            send(Message.builder()
                    .to(m.getFrom())
                    .type(MessageType.MsgAppResp)
                    .index(raftLog.lastIndex())
                    .build()
            );
        } else {
            LOGGER.info("{} [commit: {}] ignore snapshot [index: {}, term: {}]",
                    id, raftLog.getCommitted(), sindex, sterm);
            send(Message.builder()
                    .to(m.getFrom())
                    .type(MessageType.MsgAppResp)
                    .index(raftLog.getCommitted())
                    .build()
            );
        }
    }

    public boolean restore(Snapshot s) {
        if (s.getMetadata().getIndex() <= raftLog.getCommitted()) {
            return false;
        }

        if (state != StateType.StateFollower) {
            LOGGER.warn("{} attempted to restore snapshot as leader; should never happen", id);
            becomeFollower(term + 1, NONE);
            return false;
        }

        boolean found = false;
        ConfState cs = s.getMetadata().getConfState();
        List<List<Long>> temp = new ArrayList<>();
        temp.add(cs.getVoters());
        temp.add(cs.getLearners());
        temp.add(cs.getVotersOutgoing());

        for (List<Long> set : temp) {
            for (Long id : set) {
                if (id == this.id) {
                    found = true;
                    break;
                }
            }
            if (found) {
                break;
            }
        }

        if (!found) {
            LOGGER.warn("{} attempted to restore snapshot but it is not in the ConfState {}; should never happen", id, cs);
            return false;
        }

        if (raftLog.matchTerm(s.getMetadata().getIndex(), s.getMetadata().getTerm())) {
            LOGGER.info("{} [commit: {}, lastindex: {}, lastterm: {}] fast-forwarded commit to snapshot [index: {}, term: {}]",
                    id, raftLog.getCommitted(), raftLog.lastIndex(), raftLog.lastTerm(), s.getMetadata().getIndex(), s.getMetadata().getTerm());
            raftLog.commitTo(s.getMetadata().getIndex());
            return false;
        }

        raftLog.restore(s);

        prs = ProgressTracker.makeProgressTracker(prs.getMaxInflight());
        Tuple<org.raftkv.raft.tracker.Config, ProgressMap> tuple;
        try {
            tuple = ConfState.restore(new Changer(prs, raftLog.lastIndex()), cs);
        } catch (Throwable e) {
            throw new RaftException(String.format("unable to restore config %s: %s", cs, e));
        }

        org.raftkv.raft.tracker.Config cfg = tuple.getFirst();
        ProgressMap prs = tuple.getSecond();
        Utils.assertConfStatesEquivalent(cs, switchToConfig(cfg, prs));

        Progress pr = this.prs.getProgress().getProgressMap().get(id);
        pr.maybeUpdate(pr.getNext() - 1);

        LOGGER.info("{} [commit: {}, lastindex: {}, lastterm: {}] restored snapshot [index: {}, term: {}]",
                id, raftLog.getCommitted(), raftLog.lastIndex(), raftLog.lastTerm(), s.getMetadata().getIndex(), s.getMetadata().getTerm());
        return true;
    }

    public boolean promotable() {
        Progress pr = prs.getProgress().getProgressMap().get(id);
        return pr != null && !pr.isLearner() && !raftLog.hasPendingSnapshot();
    }

    public ConfState applyConfChange(ConfChangeV2 cc) {
        Supplier<Tuple<org.raftkv.raft.tracker.Config, ProgressMap>> f = () -> {
            Changer changer = new Changer(this.prs, this.raftLog.lastIndex());
            if (cc.leaveJoint()) {
                return changer.leaveJoint();
            } else {
                Tuple<Boolean, Boolean> t = cc.enterJoint();
                Boolean autoLeave = t.getFirst();
                Boolean ok = t.getSecond();
                if (ok) {
                    return changer.enterJoint(autoLeave, cc.getChanges());
                }
            }
            return changer.simple(cc.getChanges());
        };
        Tuple<org.raftkv.raft.tracker.Config, ProgressMap> tuple;
        try {
            tuple = f.get();
        } catch (Throwable e) {
            throw new RaftException(e);
        }

        return switchToConfig(tuple.getFirst(), tuple.getSecond());
    }

    private ConfState switchToConfig(org.raftkv.raft.tracker.Config cfg, ProgressMap prs) {
        this.prs.setConfig(cfg);
        this.prs.setProgress(prs);

        LOGGER.info("{} switched to configuration {}", id, this.prs.getConfig());
        ConfState cs = this.prs.confState();
        org.raftkv.raft.tracker.Progress pr = this.prs.getProgress().getProgressMap().get(id);

        this.isLearner = pr != null && pr.isLearner();

        if ((pr == null || this.isLearner) && this.state == StateType.StateLeader) {
            return cs;
        }

        if (this.state != StateType.StateLeader || cs.getVoters().size() == 0) {
            return cs;
        }

        if (maybeCommit()) {
            bcastAppend();
        } else {
            this.prs.visit((id, prs1) -> {
                maybeSendAppend(id, false);
            });
        }

        boolean tOk = this.prs.getConfig().getVoters().ids().contains(this.leadTransferee);
        if (!tOk && this.leadTransferee != 0) {
            abortLeaderTransfer();
        }

        return cs;
    }

    private void loadState(HardState state) {
        RaftLog raftLog = this.raftLog;
        if (state.getCommit() < raftLog.getCommitted() || state.getCommit() > raftLog.lastIndex()) {
            LOGGER.error("{} state.commit {} is out of range [{}, {}]",
                    id, state.getCommit(), raftLog.getCommitted(), raftLog.lastIndex());
            throw new RaftException();
        }
        raftLog.setCommitted(state.getCommit());
        this.term = state.getTerm();
        this.vote = state.getVote();
    }

    public boolean pastElectionTimeout() {
        return electionElapsed >= randomizedElectionTimeout;
    }

    public void resetRandomizedElectionTimeout() {
        randomizedElectionTimeout = electionTimeout + ThreadLocalRandom.current().nextInt(electionTimeout);
    }

    private void sendTimeoutNow(long to) {
        send(Message.builder()
                .to(to)
                .type(MessageType.MsgTimeoutNow)
                .build());
    }

    private void abortLeaderTransfer() {
        leadTransferee = NONE;
    }

    private boolean committedEntryInCurrentTerm() {
        return raftLog.zeroTermOnErrCompacted(raftLog.getCommitted()) == term;
    }

    private Message responseToReadIndexReq(Message req, long readIndex) {
        if (req.getFrom() == NONE || req.getFrom() == id) {
            readStates.add(new ReadState(readIndex, req.getEntries().get(0).getData()));
            return new Message.MessageBuilder()
                    .build();
        }
        return new Message.MessageBuilder()
                .type(MessageType.MsgReadIndexResp)
                .to(req.getFrom())
                .index(readIndex)
                .entries(req.getEntries())
                .build();
    }

    private boolean increaseUncommittedSize(List<Entry> ents) {
        long s = 0;
        for (Entry e : ents) {
            s += Utils.payloadSize(e);
        }

        if (uncommittedSize > 0 && s > 0 && uncommittedSize + s > maxUncommittedSize) {
            return false;
        }
        uncommittedSize += s;
        return true;
    }

    public void reduceUncommittedSize(List<Entry> ents) {
        if (uncommittedSize == 0) {
            return;
        }

        long s = 0;
        for (Entry e : ents) {
            s += Utils.payloadSize(e);
        }
        if (s > uncommittedSize) {
            uncommittedSize = 0;
        } else {
            uncommittedSize -= s;
        }
    }

    public long getId() {
        return id;
    }

    public void setId(long id) {
        this.id = id;
    }

    public long getTerm() {
        return term;
    }

    public void setTerm(long term) {
        this.term = term;
    }

    public RaftLog getRaftLog() {
        return raftLog;
    }

    public int getElectionElapsed() {
        return electionElapsed;
    }

    public List<ReadState> getReadStates() {
        return readStates;
    }

    public void setReadStates(List<ReadState> readStates) {
        this.readStates = readStates;
    }

    public ReadOnly getReadOnly() {
        return readOnly;
    }

    public int getHeartbeatTimeout() {
        return heartbeatTimeout;
    }


    public long getLead() {
        return lead;
    }

    public int getElectionTimeout() {
        return electionTimeout;
    }

    public void setCheckQuorum(boolean checkQuorum) {
        this.checkQuorum = checkQuorum;
    }

    public long getVote() {
        return vote;
    }

    public void setRandomizedElectionTimeout(int randomizedElectionTimeout) {
        this.randomizedElectionTimeout = randomizedElectionTimeout;
    }

    public long getMaxUncommittedSize() {
        return maxUncommittedSize;
    }

    public void setMaxUncommittedSize(long maxUncommittedSize) {
        this.maxUncommittedSize = maxUncommittedSize;
    }

    public ProgressTracker getPrs() {
        return prs;
    }

    public void setPrs(ProgressTracker prs) {
        this.prs = prs;
    }

    public void setVote(long vote) {
        this.vote = vote;
    }

    public void setRaftLog(RaftLog raftLog) {
        this.raftLog = raftLog;
    }

    public long getMaxMsgSize() {
        return maxMsgSize;
    }

    public void setMaxMsgSize(long maxMsgSize) {
        this.maxMsgSize = maxMsgSize;
    }

    public boolean isLearner() {
        return isLearner;
    }

    public void setLearner(boolean learner) {
        isLearner = learner;
    }

    public List<Message> getMsgs() {
        return msgs;
    }

    public void setMsgs(List<Message> msgs) {
        this.msgs = msgs;
    }

    public void setLead(long lead) {
        this.lead = lead;
    }

    public long getLeadTransferee() {
        return leadTransferee;
    }

    public void setLeadTransferee(long leadTransferee) {
        this.leadTransferee = leadTransferee;
    }

    public long getPendingConfIndex() {
        return pendingConfIndex;
    }

    public void setPendingConfIndex(long pendingConfIndex) {
        this.pendingConfIndex = pendingConfIndex;
    }

    public long getUncommittedSize() {
        return uncommittedSize;
    }

    public void setUncommittedSize(long uncommittedSize) {
        this.uncommittedSize = uncommittedSize;
    }

    public void setReadOnly(ReadOnly readOnly) {
        this.readOnly = readOnly;
    }

    public void setElectionElapsed(int electionElapsed) {
        this.electionElapsed = electionElapsed;
    }

    public int getHeartbeatElapsed() {
        return heartbeatElapsed;
    }

    public void setHeartbeatElapsed(int heartbeatElapsed) {
        this.heartbeatElapsed = heartbeatElapsed;
    }

    public boolean isCheckQuorum() {
        return checkQuorum;
    }

    public boolean isPreVote() {
        return preVote;
    }

    public void setPreVote(boolean preVote) {
        this.preVote = preVote;
    }

    public void setHeartbeatTimeout(int heartbeatTimeout) {
        this.heartbeatTimeout = heartbeatTimeout;
    }

    public void setElectionTimeout(int electionTimeout) {
        this.electionTimeout = electionTimeout;
    }

    public int getRandomizedElectionTimeout() {
        return randomizedElectionTimeout;
    }

    public boolean isDisableProposalForwarding() {
        return disableProposalForwarding;
    }

    public void setDisableProposalForwarding(boolean disableProposalForwarding) {
        this.disableProposalForwarding = disableProposalForwarding;
    }

    public TickFunction getTick() {
        return tick;
    }

    public void setTick(TickFunction tick) {
        this.tick = tick;
    }

    public StepFunction getStep() {
        return step;
    }

    public void setStep(StepFunction step) {
        this.step = step;
    }

    public List<Message> getPendingReadIndexMessages() {
        return pendingReadIndexMessages;
    }

    public void setPendingReadIndexMessages(List<Message> pendingReadIndexMessages) {
        this.pendingReadIndexMessages = pendingReadIndexMessages;
    }

    public StateType getState() {
        return state;
    }

    public void setState(StateType state) {
        this.state = state;
    }

    @Override
    public List<Message> readMessages() {
        List<Message> msgs = this.msgs;
        this.msgs = new ArrayList<>();

        return msgs;
    }

}
